package cn.getech.data.development.service.impl;

import cn.getech.data.development.constant.DataDevelopmentBizExceptionEnum;
import cn.getech.data.development.entity.AnalysisInfo;
import cn.getech.data.development.entity.ConfConnect;
import cn.getech.data.development.entity.ModelDetails;
import cn.getech.data.development.entity.TableFieldInfo;
import cn.getech.data.development.mapper.AnalysisInfoMapper;
import cn.getech.data.development.mapper.ConfConnectMapper;
import cn.getech.data.development.mapper.ModelDetailsMapper;
import cn.getech.data.development.mapper.TableFieldInfoMapper;
import cn.getech.data.development.model.dto.AnalysisInfoDto;
import cn.getech.data.development.model.dto.ConfConnectDto;
import cn.getech.data.development.model.qo.AnalysisInfoQO;
import cn.getech.data.development.service.IAnalysisInfoService;
import cn.getech.data.development.service.IAnalysisMessageModelConfService;
import cn.getech.data.development.utils.MyKafkaUtils;
import cn.getech.data.intelligence.common.exception.RRException;
import cn.getech.data.intelligence.common.utils.PageBean;
import cn.getech.data.intelligence.common.utils.PageUtils;
import cn.getech.data.intelligence.common.utils.PojoUtils;
import cn.getech.data.intelligence.common.utils.QueryBean;
import cn.getech.system.center.utils.ShiroUtils;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import java.time.LocalDateTime;
import java.util.List;
import java.util.Objects;

/**
 * <p>
 * 分析任务表 服务实现类
 * </p>
 *
 * @author hujz
 * @since 2019-08-21
 */
@Service
public class AnalysisInfoServiceImpl extends ServiceImpl<AnalysisInfoMapper, AnalysisInfo> implements IAnalysisInfoService {

    @Autowired
    private AnalysisInfoMapper analysisInfoMapper;


    @Autowired
    private ConfConnectMapper confConnectMapper;

    @Autowired
    private ThreadPoolTaskExecutor myExecutor;

    @Autowired
    public IAnalysisMessageModelConfService iIAnalysisMessageModelConfService;

    @Autowired
    private TableFieldInfoMapper tableFieldInfoMapper;

    @Autowired
    private ModelDetailsMapper modelDetailsMapper;

    @Override
    public void deleteAnalysisInfo(String id) {
        //其下有启动的分析任务，则不能删除
        if (StringUtils.isEmpty(id)) {
            return;
        }
        AnalysisInfoDto analysisInfoDto = editById(id);
        if (null == analysisInfoDto) {
            throw new RRException(String.format(DataDevelopmentBizExceptionEnum.PARAM_NO_EXIT_ERROR.getMessage(), "该分析任务"));
        }
        if (analysisInfoDto.getEnable()) {
            throw new RRException(String.format(DataDevelopmentBizExceptionEnum.ANALYSIS_INFO_STATUS_FAIL.getMessage(), analysisInfoDto.getEnable() ? "启用" : "禁用"));
        }
        analysisInfoMapper.deleteById(id);

    }

    @Override
    public void save(AnalysisInfoDto analysisInfoDto) {
        //验证名称重复
        if (null == analysisInfoDto || StringUtils.isEmpty(analysisInfoDto.getName())) {
            throw new RRException(String.format(DataDevelopmentBizExceptionEnum.PARAM_NEED_FAIL_ERROR.getMessage(), "分析任务的名称"));
        }
        if (null == analysisInfoDto.getProcId()) {
            throw new RRException(String.format(DataDevelopmentBizExceptionEnum.PARAM_NEED_FAIL_ERROR.getMessage(), "工程id"));
        }
        Boolean aBoolean = checkAnalysisName(analysisInfoDto.getId(), analysisInfoDto.getName());
        if (aBoolean) {
            throw new RRException(String.format(DataDevelopmentBizExceptionEnum.NAME_DOUBLE_FAIL_ERROR.getMessage(), "分析任务名称"));
        }
        if (null != analysisInfoDto.getId()) {
            //修改
            AnalysisInfo analysisInfo = analysisInfoMapper.selectById(analysisInfoDto.getId());
            if (null == analysisInfo) {
                throw new RRException(String.format(DataDevelopmentBizExceptionEnum.PARAM_NO_EXIT_ERROR.getMessage(), "分析任务"));
            }
            if (analysisInfo.getEnable()) {
                throw new RRException(String.format(DataDevelopmentBizExceptionEnum.ANALYSIS_INFO_STATUS_FAIL.getMessage(), analysisInfo.getEnable() ? "启用" : "禁用"));
            }
            analysisInfo.setName(analysisInfoDto.getName());
            analysisInfo.setDescption(analysisInfoDto.getDescption());
            analysisInfo.setModelInfoId(ShiroUtils.getUserId().intValue());
            analysisInfo.setModTime(LocalDateTime.now());
            analysisInfoMapper.updateById(analysisInfo);
        } else {
            //新增
            AnalysisInfo analysisInfo = new AnalysisInfo();
            BeanUtils.copyProperties(analysisInfoDto, analysisInfo);
            analysisInfo.setEnable(false);
            analysisInfo.setCreatePer(ShiroUtils.getUserId().intValue());
            analysisInfo.setCreateTime(LocalDateTime.now());
            analysisInfoMapper.insert(analysisInfo);
        }

    }

    /**
     * @Description 验证分析任务的名称的重复问题
     * @Author hujz
     * @Date 2019/8/22 11:47
     * @Param
     * @Return 是否重复
     * @Exception
     */
    private Boolean checkAnalysisName(Integer id, String name) {
        if (StringUtils.isEmpty(name)) {
            return true;
        }
        Integer count = analysisInfoMapper.selectCount(new QueryWrapper<AnalysisInfo>().eq("name", name).notIn(null != id, "id", id));
        return count > 0;
    }

    @Override
    public AnalysisInfoDto editById(String id) {
        if (StringUtils.isEmpty(id)) {
            return null;
        }
        AnalysisInfo analysisInfo = analysisInfoMapper.selectById(id);
        if (null != analysisInfo) {
            AnalysisInfoDto analysisInfoDto = new AnalysisInfoDto();
            BeanUtils.copyProperties(analysisInfo, analysisInfoDto);
            return analysisInfoDto;
        }
        return null;
    }

    @Override
    public void update(AnalysisInfoDto analysisInfoDto) {
        if (null == analysisInfoDto || null == analysisInfoDto.getId()) {
            throw new RRException(String.format(DataDevelopmentBizExceptionEnum.PARAM_NEED_FAIL_ERROR.getMessage(), "分析任务的id"));
        }
        AnalysisInfo analysisInfo = analysisInfoMapper.selectById(analysisInfoDto.getId());
        if (null == analysisInfo) {
            throw new RRException(String.format(DataDevelopmentBizExceptionEnum.PARAM_NO_EXIT_ERROR.getMessage(), "分析任务"));
        }
        if (analysisInfo.getEnable()) {
            throw new RRException(String.format(DataDevelopmentBizExceptionEnum.ANALYSIS_INFO_STATUS_FAIL.getMessage(), analysisInfo.getEnable() ? "启用" : "禁用"));
        }
        //验证名称重复
        if (StringUtils.isEmpty(analysisInfoDto.getName())) {
            throw new RRException(String.format(DataDevelopmentBizExceptionEnum.PARAM_NEED_FAIL_ERROR.getMessage(), "分析任务的名称"));
        }
        Boolean aBoolean = checkAnalysisName(analysisInfoDto.getId(), analysisInfoDto.getName());
        if (aBoolean) {
            throw new RRException(String.format(DataDevelopmentBizExceptionEnum.NAME_DOUBLE_FAIL_ERROR.getMessage(), "分析任务名称"));
        }
        analysisInfo.setName(analysisInfoDto.getName());
        analysisInfo.setDescption(analysisInfoDto.getDescption());
        analysisInfo.setModelInfoId(ShiroUtils.getUserId().intValue());
        analysisInfo.setModTime(LocalDateTime.now());
        analysisInfoMapper.updateById(analysisInfo);

    }

    @Override
    public List<AnalysisInfoDto> list(AnalysisInfoQO analysisInfoQO) {
        //按照名称模糊查询
        if (null == analysisInfoQO || null == analysisInfoQO.getProcId()) {
            throw new RRException(String.format(DataDevelopmentBizExceptionEnum.PARAM_NEED_FAIL_ERROR.getMessage(), "工程id"));
        }
        String analysisName = "";
        if (StringUtils.isNotEmpty(analysisInfoQO.getName())) {
            analysisName = analysisInfoQO.getName();
        }
        List<AnalysisInfo> analysisInfos = analysisInfoMapper.selectList(new QueryWrapper<AnalysisInfo>()
                .eq("create_per", ShiroUtils.getUserId().intValue())
                .eq("proc_id", analysisInfoQO.getProcId())
                .like("name", analysisName));
        return PojoUtils.listConvert(AnalysisInfoDto.class, analysisInfos);
    }

    @Override
    public PageUtils page(AnalysisInfoQO analysisInfoQO) {
        //按照名称模糊分页查询
        if (null == analysisInfoQO || null == analysisInfoQO.getProcId()) {
            throw new RRException(String.format(DataDevelopmentBizExceptionEnum.PARAM_NEED_FAIL_ERROR.getMessage(), "工程id"));
        }
        String analysisName = "";
        if (StringUtils.isNotEmpty(analysisInfoQO.getName())) {
            analysisName = analysisInfoQO.getName();
        }
        IPage<AnalysisInfo> infoIPage = analysisInfoMapper.selectPage(new QueryBean<AnalysisInfo, PageBean>().getPage(analysisInfoQO)
                , new QueryWrapper<AnalysisInfo>()
                        .eq("create_per", ShiroUtils.getUserId().intValue())
                        .eq("proc_id", analysisInfoQO.getProcId())
                        .like("name", analysisName));

        return new PageUtils(infoIPage);
    }

    @Override
    public void updateEnable(AnalysisInfoDto analysisInfoDto) {
        if (null == analysisInfoDto || null == analysisInfoDto.getId()) {
            throw new RRException(String.format(DataDevelopmentBizExceptionEnum.PARAM_NEED_FAIL_ERROR.getMessage(), "分析任务的id"));
        }
        if (null == analysisInfoDto.getEnable()) {
            throw new RRException(String.format(DataDevelopmentBizExceptionEnum.PARAM_NEED_FAIL_ERROR.getMessage(), "分析任务的状态"));
        }
        AnalysisInfo analysisInfo = analysisInfoMapper.selectById(analysisInfoDto.getId());
        if (null == analysisInfo) {
            throw new RRException(String.format(DataDevelopmentBizExceptionEnum.PARAM_NO_EXIT_ERROR.getMessage(), "分析任务"));
        }
        if (Objects.equals(analysisInfo.getEnable(), analysisInfoDto.getEnable())) {
            throw new RRException("状态一致，无需操作！");
        }
        // 还得处理相关的操作，不再处理数据源中的数据等等.禁用：则需要处理数据。启用需要验证数据，并且去跑模型
        // 执行flink的命令  ./bin/flink run   -c cn.getech.data.development.job.flink.RunningModel  /var/lib/hadoop-hdfs/flink-demo/bdp-common-1.0.0-jar-with-dependencies.jar
        // 停止flink的命令  ./bin/flink stop [-p targetDirectory] [-d] <jobID>


        analysisInfo.setEnable(analysisInfoDto.getEnable());
        analysisInfoMapper.updateById(analysisInfo);
    }

    @Override
    public void updateUp(Integer id) {
        if (null == id) {
            throw new RRException(String.format(DataDevelopmentBizExceptionEnum.PARAM_NEED_FAIL_ERROR.getMessage(), "分析任务的id"));
        }
        AnalysisInfo analysisInfo = analysisInfoMapper.selectById(id);
        if (null == analysisInfo) {
            throw new RRException(String.format(DataDevelopmentBizExceptionEnum.PARAM_NO_EXIT_ERROR.getMessage(), "分析任务"));
        }
        if (Objects.equals(analysisInfo.getEnable(), true)) {
            throw new RRException("分析任务的状态已为启用");
        }
        checkAllData(analysisInfo);
        ConfConnectDto confConnectDto = confConnectMapper.selectAllanalysisInfoDataStatusUpStop(analysisInfo.getId().toString());
        if (null != confConnectDto) {
            myExecutor.execute(new Runnable() {
                @Override
                public void run() {
                    String uuids = MyKafkaUtils.initKafKaConsumer(confConnectDto);
                    String key = uuids + "_" + analysisInfo.getConnectId();
//                    kafkaTestModelConsumer.startModel(key);
                }
            });
        }
        analysisInfo.setEnable(true);
        analysisInfoMapper.updateById(analysisInfo);
    }

    /***
     * 检查配置的分析任务是否正确
     * @param analysisInfo 分析任务信息
     */
    private void checkAllData(AnalysisInfo analysisInfo) {
        if (null == analysisInfo) {
            return;
        }
        //验证配置的对应的id信息
        if (null == analysisInfo.getConnectId() || null == analysisInfo.getModelInfoId() || null == analysisInfo.getTableInfoId()) {
            throw new RRException("分析任务配置不正确");
        }
        //验证连接的参数
        ConfConnect confConnect = confConnectMapper.selectById(analysisInfo.getConnectId());
        iIAnalysisMessageModelConfService.cheackConfConnect(confConnect);

        //验证模型的参数
        //验证其输出字段是否一致
        List<TableFieldInfo> tableFieldInfos = tableFieldInfoMapper.selectList(new QueryWrapper<TableFieldInfo>().eq("table_id", analysisInfo.getTableInfoId()));
        if (null == tableFieldInfos || tableFieldInfos.isEmpty()) {
            throw new RRException("输出仓库不存在,输出字段！");
        }
        List<ModelDetails> modelDetailsOutPut = modelDetailsMapper.selectList(new QueryWrapper<ModelDetails>().eq("model_type", 2).eq("model_info_id", analysisInfo.getModelInfoId()));
        if (null == modelDetailsOutPut || modelDetailsOutPut.isEmpty()) {
            throw new RRException("模型没有输出字段！");
        }
        iIAnalysisMessageModelConfService.cheackModelField(tableFieldInfos, modelDetailsOutPut);

    }

    @Override
    public void updateDown(Integer id) {
        if (null == id) {
            throw new RRException(String.format(DataDevelopmentBizExceptionEnum.PARAM_NEED_FAIL_ERROR.getMessage(), "分析任务的id"));
        }
        AnalysisInfo analysisInfo = analysisInfoMapper.selectById(id);
        if (null == analysisInfo) {
            throw new RRException(String.format(DataDevelopmentBizExceptionEnum.PARAM_NO_EXIT_ERROR.getMessage(), "分析任务"));
        }
        if (Objects.equals(analysisInfo.getEnable(), false)) {
            throw new RRException("分析任务的状态已为禁用");
        }
        myExecutor.execute(new Runnable() {
            @Override
            public void run() {
                String key = analysisInfo.getId() + "_" + analysisInfo.getConnectId();
//                kafkaTestModelConsumer.stopRunningModel(key);
            }
        });
        analysisInfo.setEnable(false);
        analysisInfoMapper.updateById(analysisInfo);

    }
}
